package operator.combine;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Function;
import util.RxJavaHelper;
import util.ThreadHelper;

class CombineLatestOperator {

    public static void main(String[] args) {
        combineLatest();
    }

    //两个Observable的数据项合并 取每个Observeble当前最后发射的数据项进行合并
    //再把合并后的结果给消费者
    static void combineLatest(){
        Observable<String> o1 = Observable.intervalRange(1, 10, 0, 1, TimeUnit.SECONDS)
        .map(new Function<Long, String>() {
            @Override
            public String apply(Long aLong) throws Exception {
                return "A " + aLong;
            }
        });
        Observable<String> o2 = Observable.intervalRange(1, 10, 3, 1, TimeUnit.SECONDS)
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "B " + aLong;
                    }
                });
        Observable.combineLatest(o1, o2, new BiFunction<String, String, String>() {
            @Override
            public String apply(String s1, String s2) throws Exception {
                return s1 + " and " + s2;
            }
        }).subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

}
